package com.hugedata.idc.etl;

import java.io.DataInputStream;
import java.io.IOException;

import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.log4j.Logger;

import com.hugedata.idc.utils.HDFSUtils;

public class DnsDelaySendHDFS
{
    private static final Logger LOGGER = Logger.getLogger(DnsDelaySendHDFS.class);

    public static void main(String[] args) throws IOException
    {
        CreateQosSendFiles(args[0], args[1]);
    }

    public static void CreateQosSendFiles(String srcDir, String destDir) throws IOException
    {
        Configuration config = HDFSUtils.getConf();
        FileSystem fs = FileSystem.get(config);
        Path srcPath = new Path(srcDir);

        FileStatus[] isps = fs.listStatus(srcPath);

        for (int j = 0; j < isps.length; j++)
        {
            FileStatus isp = isps[j];

            String ispname = isp.getPath().getName();

            String lp = ispname.substring(4);

            if (isp.isDirectory())
            {

                String filename = lp + ".txt";

                FileStatus[] files = fs.listStatus(isp.getPath());

                for (int k = 0; k < files.length; k++)
                {

                    if (files[k].isFile())
                    {
                        Path destPath = new Path(destDir, filename);

                        DataInputStream inputStream = fs.open(files[k].getPath());

                        FSDataOutputStream outputStream;
                        if (fs.exists(destPath))
                        {
                            outputStream = fs.append(destPath);
                        }
                        else
                        {
                            outputStream = fs.create(destPath);
                        }
                        IOUtils.copyLarge(inputStream, outputStream);
                        IOUtils.closeQuietly(inputStream);
                        IOUtils.closeQuietly(outputStream);
                    }
                }
            }
        }
    }
}
